package com.yuelao.hbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.TableDescriptor;

import java.io.IOException;

/**
 * Description
 *
 * @author fengliye
 * @version v1.0
 * @date 28/12/2018 18:21
 * @modify detail
 * 版本       修改人        修改时间      修改内容描述
 * ---------------------------------------------
 * <p>
 * ---------------------------------------------
 */
@Slf4j
public class HBaseClient {
    private static Configuration configuration = null;
    private static Connection connection = null;
    private static Admin HB_ADMIN = null;


    private static class SingletonHolder {
        private static final HBaseClient INSTANCE = new HBaseClient();
    }

    private HBaseClient() {
        try {
            configuration = HBaseConfiguration.create();
            configuration.set("zookeeper.znode.parent", "/hbase/master"); //与 hbase-site-xml里面的配置信息 zookeeper.znode.parent 一致
            configuration.set("hbase.zookeeper.quorum", "localhost");  //hbase 服务地址
            configuration.set("hbase.zookeeper.property.clientPort", "2181"); //端口号
            connection = ConnectionFactory.createConnection(configuration);
        } catch (Exception e) {
            log.error("init hbase connect error: {}", e);
        }
    }

    public static final Connection getConn() {
        return SingletonHolder.INSTANCE.connection;
    }

    public static final Admin getAdmin() {
        if (HB_ADMIN == null) {
            try {
                HB_ADMIN = getConn().getAdmin();
            } catch (IOException e) {
                log.error("init admin error: {}", e);
            }
        }

        return HB_ADMIN;
    }

    public static final Configuration getConf() {
        getConn();
        return configuration;
    }

    /**
     * @param descriptor
     * @return
     */
    public static boolean createTable(TableDescriptor descriptor) {
        //To String convert to TableName
        TableName table = TableName.valueOf(descriptor.getTableName().getNameAsString());
        Boolean flag = false;
        try {
            //判断tname是否存在,存在就返回true,否则返回false
            flag = getAdmin().tableExists(table);
            if (flag) {
                log.error("Table has existed");
                return !flag;
            } else {
                //判断当前的表是否被禁用了,是就开启
                if (getAdmin().isTableDisabled(table)) {
                    getAdmin().enableTable(table);
                }

                getAdmin().createTable(descriptor);
            }
        } catch (IOException e) {
            log.error("Table create error : {}", e);
        }
        return !flag;

    }

    public Configuration getConfiguration() {
        return this.configuration;
    }

}
